/**
 * Copyright 2018 VMware, Inc.
 * <p>
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 * <p>
 * https://www.apache.org/licenses/LICENSE-2.0
 * <p>
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package cn.com.ry.framework.monitormeter.instrument.linklog;

import com.alibaba.fastjson.JSON;
import io.micrometer.core.annotation.Incubating;
import io.micrometer.core.instrument.*;
import io.micrometer.core.instrument.binder.BaseUnits;
import io.micrometer.core.instrument.config.NamingConvention;
import io.micrometer.core.instrument.distribution.DistributionStatisticConfig;
import io.micrometer.core.instrument.distribution.HistogramSnapshot;
import io.micrometer.core.instrument.distribution.pause.PauseDetector;
import io.micrometer.core.instrument.step.StepDistributionSummary;
import io.micrometer.core.instrument.step.StepMeterRegistry;
import io.micrometer.core.instrument.step.StepTimer;
import io.micrometer.core.instrument.util.NamedThreadFactory;
import io.micrometer.core.lang.Nullable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;
import java.util.stream.StreamSupport;

import static io.micrometer.core.instrument.util.DoubleFormat.decimalOrNan;
import static java.util.stream.Collectors.joining;

/**
 * @author reywong
 */
@Incubating(since = "1.1.0")
public class LinkLogMeterRegistry extends StepMeterRegistry {
    private static final Logger log = LoggerFactory.getLogger(LinkLogMeterRegistry.class);

    private final LinkLogRegistryConfig config;
    private final ConcurrentMap linkLogSink;

    public LinkLogMeterRegistry() {
        this(LinkLogRegistryConfig.DEFAULT, Clock.SYSTEM);
    }

    public LinkLogMeterRegistry(LinkLogRegistryConfig config, Clock clock) {
        this(config, clock, new NamedThreadFactory("linkLog-metrics-publisher"), new ConcurrentHashMap(), null);
    }

    private LinkLogMeterRegistry(LinkLogRegistryConfig config, Clock clock, ThreadFactory threadFactory, ConcurrentMap linkLogSink, @Nullable Function<Meter, String> meterIdPrinter) {
        super(config, clock);
        this.config = config;
        this.linkLogSink = linkLogSink;
        config().namingConvention(NamingConvention.dot);
        start(threadFactory);
    }

    @Override
    protected void publish() {
        if (config.enabled()) {
            List<Meter> meterList = getMeters();

            if (meterList != null && meterList.size() > 0) {
                for (Meter meter : meterList) {
                    Printer print = new Printer(meter);
                    meter.use(gauge -> linkLogSink.put(print.id() + ".value", print.value(gauge.value())),
                            counter -> {
                                double count = counter.count();
                                if (!config.logInactive() && count == 0) {
                                    return;
                                }
                                linkLogSink.put(print.id() + ".throughput", print.rate(count));
                            },
                            timer -> {
                                HistogramSnapshot snapshot = timer.takeSnapshot();
                                long count = snapshot.count();
                                if (!config.logInactive() && count == 0) {
                                    return;
                                }

                                linkLogSink.put(print.id() + ".throughput", print.unitlessRate(count));
                                linkLogSink.put(print.id() + ".mean", print.time(snapshot.mean(getBaseTimeUnit())));
                                linkLogSink.put(print.id() + ".max", print.time(snapshot.max(getBaseTimeUnit())));
                            },
                            summary -> {
                                HistogramSnapshot snapshot = summary.takeSnapshot();
                                long count = snapshot.count();
                                if (!config.logInactive() && count == 0) {
                                    return;
                                }
                                linkLogSink.put(print.id() + ".throughput", print.unitlessRate(count));
                                linkLogSink.put(print.id() + ".mean", print.value(snapshot.mean()));
                                linkLogSink.put(print.id() + ".max", print.value(snapshot.max()));
                            },
                            longTaskTimer -> {
                                int activeTasks = longTaskTimer.activeTasks();
                                if (!config.logInactive() && activeTasks == 0) {
                                    return;
                                }
                                //获取tag
                                List<Tag> tagList = longTaskTimer.getId().getTags();
                                String clazz = "";
                                String method = "";
                                if (tagList != null) {
                                    for (Tag tag : tagList) {
                                        if (tag != null) {
                                            if (tag.getKey().equalsIgnoreCase("method")) {
                                                method = tag.getValue();
                                            }

                                            if (tag.getKey().equalsIgnoreCase("class")) {
                                                clazz = tag.getValue();
                                                if (clazz != null) {
                                                    String[] clazzs = clazz.split("\\.");
                                                    if (clazzs != null && clazzs.length > 0) {
                                                        clazz = clazzs[clazzs.length - 1];
                                                    }

                                                }
                                            }
                                        }
                                    }
                                }
                                String timerKey = "";
                                if (!clazz.equalsIgnoreCase("")) {
                                    timerKey += "." + clazz;
                                }
                                if (!method.equalsIgnoreCase("")) {
                                    timerKey += "." + method;
                                }
                                linkLogSink.put(print.id() + timerKey + ".active", print.value(activeTasks));
                                linkLogSink.put(print.id() + timerKey + ".duration", print.time(longTaskTimer.duration(getBaseTimeUnit())));
                            },
                            timeGauge -> {
                                double value = timeGauge.value(getBaseTimeUnit());
                                if (!config.logInactive() && value == 0) {
                                    return;
                                }
                                linkLogSink.put(print.id() + ".value", print.time(value));
                            },
                            counter -> {
                                double count = counter.count();
                                if (!config.logInactive() && count == 0) {
                                    return;
                                }
                                linkLogSink.put(print.id() + ". throughput", print.rate(count));
                            },
                            timer -> {
                                double count = timer.count();
                                if (!config.logInactive() && count == 0) {
                                    return;
                                }
                                linkLogSink.put(print.id() + ". throughput", print.rate(count));
                                linkLogSink.put(print.id() + ". mean", print.time(timer.mean(getBaseTimeUnit())));
                            },
                            visitMeter -> linkLogSink.put(print.id() + ".value", writeMeter(visitMeter, print))
                    );
                }

                log.info(JSON.toJSONString(linkLogSink));
            }

        }
    }

    String writeMeter(Meter meter, Printer print) {
        return StreamSupport.stream(meter.measure().spliterator(), false)
                .map(ms -> {
                    String msLine = ms.getStatistic().getTagValueRepresentation() + "=";
                    switch (ms.getStatistic()) {
                        case TOTAL:
                        case MAX:
                        case VALUE:
                            return msLine + print.value(ms.getValue());
                        case TOTAL_TIME:
                        case DURATION:
                            return msLine + print.time(ms.getValue());
                        case COUNT:
                            return "throughput=" + print.rate(ms.getValue());
                        default:
                            return msLine + decimalOrNan(ms.getValue());
                    }
                })
                .collect(joining());
    }

    @Override
    protected Timer newTimer(Meter.Id id, DistributionStatisticConfig distributionStatisticConfig, PauseDetector pauseDetector) {
        return new StepTimer(id, clock, distributionStatisticConfig, pauseDetector, getBaseTimeUnit(),
                this.config.step().toMillis(), false);
    }

    @Override
    protected DistributionSummary newDistributionSummary(Meter.Id id, DistributionStatisticConfig distributionStatisticConfig, double scale) {
        return new StepDistributionSummary(id, clock, distributionStatisticConfig, scale,
                config.step().toMillis(), false);
    }

    class Printer {
        private final Meter meter;

        Printer(Meter meter) {
            this.meter = meter;
        }

        String id() {
            return meter.getId().getName();
        }

        double time(double time) {
            return time / 1000.0;
        }

        double rate(double rate) {
            return rate / (double) config.step().getSeconds();
        }

        double unitlessRate(double rate) {
            return rate / (double) config.step().getSeconds();
        }

        double value(double value) {
            return value;
        }

        // see https://stackoverflow.com/a/3758880/510017
        String humanReadableByteCount(double bytes) {
            int unit = 1024;
            if (bytes < unit || Double.isNaN(bytes)) {
                return decimalOrNan(bytes) + " B";
            }
            int exp = (int) (Math.log(bytes) / Math.log(unit));
            String pre = "KMGTPE".charAt(exp - 1) + "i";
            return decimalOrNan(bytes / Math.pow(unit, exp)) + " " + pre + "B";
        }

        String humanReadableBaseUnit(double value) {
            String baseUnit = meter.getId().getBaseUnit();
            if (BaseUnits.BYTES.equals(baseUnit)) {
                return humanReadableByteCount(value);
            }
            return decimalOrNan(value) + (baseUnit != null ? " " + baseUnit : "");
        }
    }

    @Override
    protected TimeUnit getBaseTimeUnit() {
        return TimeUnit.MILLISECONDS;
    }

    public static Builder builder(LinkLogRegistryConfig config) {
        return new Builder(config);
    }

    public static class Builder {
        private final LinkLogRegistryConfig config;

        private Clock clock = Clock.SYSTEM;
        private ThreadFactory threadFactory = new NamedThreadFactory("linkLog-metrics-publisher");
        private ConcurrentMap loggingSink = new ConcurrentHashMap();
        @Nullable
        private Function<Meter, String> meterIdPrinter;

        Builder(LinkLogRegistryConfig config) {
            this.config = config;
        }

        public Builder clock(Clock clock) {
            this.clock = clock;
            return this;
        }

        public Builder threadFactory(ThreadFactory threadFactory) {
            this.threadFactory = threadFactory;
            return this;
        }

        public Builder loggingSink(ConcurrentMap loggingSink) {
            this.loggingSink = loggingSink;
            return this;
        }

        /**
         * Configure printer for meter IDs.
         *
         * @param meterIdPrinter printer to use for meter IDs
         * @return this builder instance
         * @since 1.2.0
         */
        public Builder meterIdPrinter(Function<Meter, String> meterIdPrinter) {
            this.meterIdPrinter = meterIdPrinter;
            return this;
        }

        public LinkLogMeterRegistry build() {
            return new LinkLogMeterRegistry(config, clock, threadFactory, loggingSink, meterIdPrinter);
        }
    }
}
